package com.lm.paymentcanal.user;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.lm.paymentcanal.bean.Account;
import com.lm.paymentcanal.common.CanalDataHandler;
import com.lm.paymentcanal.kafka.KafkaProducer;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

@Component
public class UserAnalysis {

    @Resource
    private KafkaProducer kafkaProducer;

    public static final String TOPIC_TEST = "account";


    public void analyseUser(RowChange rowChange){
        for (RowData rowData : rowChange.getRowDatasList()) {
            if (rowChange.getEventType() == EventType.UPDATE) {
                //用户account变化
                Account before = CanalDataHandler.convertToBean(rowData.getBeforeColumnsList(),Account.class);
                Account after = CanalDataHandler.convertToBean(rowData.getAfterColumnsList(),Account.class);
                JSONObject object = new JSONObject();
                object.put("account_id",after.getAccount_id());
                object.put("user_id",after.getUser_id());
                object.put("money",after.getMoney().subtract(before.getMoney()).doubleValue());
                object.put("time_dif",after.getUpdate_time().getTime()-before.getUpdate_time().getTime());
                Date d = new Date(after.getUpdate_time().getTime());
                SimpleDateFormat sbf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                object.put("ts",sbf.format(d));
                kafkaProducer.sendMessage(TOPIC_TEST,object);
            }
        }
    }


}
